7.3　work

work 包的目的是展示如何使用无缓冲的通道来创建一个goroutine池，这些goroutine执行并控制一组工作，让其并发执行。在这种情况下，使用无缓冲的通道要比随意指定一个缓冲区大小的有缓冲的通道好，因为这个情况下既不需要一个工作队列，也不需要一组goroutine配合执行。无缓冲的通道保证两个goroutine之间的数据交换。这种使用无缓冲的通道的方法允许使用者知道什么时候goroutine池正在执行工作，而且如果池里的所有goroutine都忙，无法接受新的工作的时候，也能及时通过通道来通知调用者。使用无缓冲的通道不会有工作在队列里丢失或者卡住，所有工作都会被处理。

让我们来看一下 work 包里的work.go代码文件，如代码清单7-28所示。

代码清单7-28　 work /work.go

01 // Jason Waldrip协助完成了这个示例
02 // work包管理一个goroutine池来完成工作
03 package work
04
05 import "sync"
06
07 // Worker必须满足接口类型，
08 // 才能使用工作池
09 type Worker interface {
10　　 Task()
11 }
12
13 // Pool提供一个goroutine池，这个池可以完成
14 // 任何已提交的Worker任务
15 type Pool struct {
16　　 work chan Worker
17　　 wg　 sync.WaitGroup
18 }
19
20 // New创建一个新工作池
21 func New(maxGoroutines int) *Pool {
22　　 p := Pool{
23　　　　 work: make(chan Worker),
24　　 }
25
26　　 p.wg.Add(maxGoroutines)
27　　 for i := 0; i < maxGoroutines; i++ {
28　　　　 go func() {
29　　　　　　 for w := range p.work {
30　　　　　　　　 w.Task()
31　　　　　　 }
32　　　　　　 p.wg.Done()
33　　　　 }()
34　　 }
35
36　　 return &p
37 }
38
39 // Run提交工作到工作池
40 func (p *Pool) Run(w Worker) {
41　　 p.work <- w
42 }
43
44 // Shutdown等待所有goroutine停止工作
45 func (p *Pool) Shutdown() {
46　　 close(p.work)
47　　 p.wg.Wait()
48 }

代码清单7-28中展示的 work 包一开始声明了名为 Worker 的接口和名为 Pool 的结构，如代码清单7-29所示。

代码清单7-29　 work /work.go：第07行到第18行

07 // Worker必须满足接口类型， 
08 // 才能使用工作池
09 type Worker interface {
10　　 Task()
11 }
12
13 // Pool提供一个goroutine池，这个池可以完成
14 // 任何已提交的Worker任务
15 type Pool struct {
16　　 work chan Worker
17　　 wg　 sync.WaitGroup
18 }

代码清单7-29的第09行中的 Worker 接口声明了一个名为 Task 的方法。在第15行，声明了名为 Pool 的结构，这个结构类型实现了goroutine池，并实现了一些处理工作的方法。这个结构类型声明了两个字段，一个名为 work （一个 Worker 接口类型的通道），另一个名为 wg 的 sync.WaitGroup 类型。

接下来，让我们来看一下 work 包的工厂函数，如代码清单7-30所示。

代码清单7-30　 work /work.go：第20行到第37行

20 // New创建一个新工作池
21 func New(maxGoroutines int) *Pool {
22　　 p := Pool{
23　　　　 work: make(chan Worker),
24　　 }
25
26　　 p.wg.Add(maxGoroutines)
27　　 for i := 0; i < maxGoroutines; i++ {
28　　　　 go func() {
29　　　　　　 for w := range p.work {
30　　　　　　　　 w.Task()
31　　　　　　 }
32　　　　　　 p.wg.Done()
33　　　　 }()
34　　 }
35
36　　 return &p
37 }

代码清单7-30展示了 New 函数，这个函数使用固定数量的goroutine来创建一个工作池。goroutine的数量作为参数传给 New 函数。在第22行，创建了一个 Pool 类型的值，并使用无缓冲的通道来初始化 work 字段。

之后，在第26行，初始化 WaitGroup 需要等待的数量，并在第27行到第34行，创建了同样数量的goroutine。这些goroutine只接收 Worker 类型的接口值，并调用这个值的 Task 方法，如代码清单7-31所示。

代码清单7-31　 work /work.go：第28行到第33行

28　　　　 go func() {
29　　　　　　 for w := range p.work {
30　　　　　　　　 w.Task()
31　　　　　　 }
32　　　　　　 p.wg.Done()
33　　　　 }()

代码清单7-31里的 for range 循环会一直阻塞，直到从 work 通道收到一个 Worker 接口值。如果收到一个值，就会执行这个值的 Task 方法。一旦 work 通道被关闭， for range 循环就会结束，并调用 WaitGroup 的 Done 方法。然后goroutine终止。

现在我们可以创建一个等待并执行工作的goroutine池了。让我们看一下如何向池里提交工作，如代码清单7-32所示。

代码清单7-32　 work /work.go：第39行到第42行

39 // Run提交工作到工作池
40 func (p *Pool) Run(w Worker) {
41　　 p.work <- w
42 }

代码清单7-32展示了 Run 方法。这个方法可以向池里提交工作。该方法接受一个 Worker 类型的接口值作为参数，并将这个值通过 work 通道发送。由于 work 通道是一个无缓冲的通道，调用者必须等待工作池里的某个goroutine接收到这个值才会返回。这正是我们想要的，这样可以保证调用的 Run 返回时，提交的工作已经开始执行。

在某个时间点，需要关闭工作池。这是 Shutdown 方法所做的事情，如代码清单7-33所示。

代码清单7-33　 work /work.go：第44行到第48行

44 // Shutdown等待所有goroutine停止工作
45 func (p *Pool) Shutdown() {
46　　 close(p.work)
47　　 p.wg.Wait()
48 }

代码清单7-33中的 Shutdown 方法做了两件事，首先，它关闭了 work 通道，这会导致所有池里的goroutine停止工作，并调用 WaitGroup 的 Done 方法；然后， Shutdown 方法调用 WaitGroup 的 Wait 方法，这会让 Shutdown 方法等待所有goroutine终止。

我们看了 work 包的代码，并了解了它是如何工作的，接下来让我们看一下main.go源代码文件中的测试程序，如代码清单7-34所示。

代码清单7-34　 work /main/main.go

01 // 这个示例程序展示如何使用work包
02 // 创建一个goroutine池并完成工作
03 package main
04
05 import (
06　　 "log"
07　　 "sync"
08　　 "time"
09
10　　 "github.com/goinaction/code/chapter7/patterns/work"
11 )
12
13 // names提供了一组用来显示的名字
14 var names = []string{
15　　 "steve",
16　　 "bob",
17　　 "mary",
18　　 "therese",
19　　 "jason",
20 }
21
22 // namePrinter使用特定方式打印名字
23 type namePrinter struct {
24　　 name string
25 }
26
27 // Task实现Worker接口
28 func (m *namePrinter) Task() {
29　　 log.Println(m.name)
30　　 time.Sleep(time.Second)
31 }
32
33 // main是所有Go程序的入口
34 func main() {
35　　 // 使用两个goroutine来创建工作池
36　　 p := work.New(2)
37
38　　 var wg sync.WaitGroup
39　　 wg.Add(100 * len(names))
40
41　　 for i := 0; i < 100; i++ {
42　　　　 // 迭代names切片
43　　　　 for _, name := range names {
44　　　　　　 // 创建一个namePrinter并提供
45　　　　　　 // 指定的名字
46　　　　　　 np := namePrinter{
47　　　　　　　　 name: name,
48　　　　　　 }
49
50　　　　　　 go func() {
51　　　　　　　　 // 将任务提交执行。当Run返回时
52　　　　　　　　 // 我们就知道任务已经处理完成
53　　　　　　　　 p.Run(&np)
54　　　　　　　　 wg.Done()
55　　　　　　 }()
56　　　　 }
57　　 }
58
59　　 wg.Wait()
60
61　　 // 让工作池停止工作，等待所有现有的
62　　 // 工作完成
63　　 p.Shutdown()
64 }

代码清单7-34展示了使用 work 包来完成名字显示工作的测试程序。这段代码一开始在第14行声明了名为 names 的包级的变量，这个变量被声明为一个字符串切片。这个切片使用5个名字进行了初始化。然后声明了名为 namePrinter 的类型，如代码清单7-35所示。

代码清单7-35　 work /main/main.go：第22行到第31行

22 // namePrinter使用特定方式打印名字
23 type namePrinter struct {
24　　 name string
25 }
26
27 // Task实现Worker接口
28 func (m *namePrinter) Task() {
29　　 log.Println(m.name)
30　　 time.Sleep(time.Second)
31 }

在代码清单7-35的第23行，声明了 namePrinter 类型，接着是这个类型对 Worker 接口的实现。这个类型的工作任务是在显示器上显示名字。这个类型只包含一个字段，即 name ，它包含要显示的名字。 Worker 接口的实现 Task 函数用 log.Println 函数来显示名字，之后等待1 秒再退出。等待这1秒只是为了让测试程序运行的速度慢一些，以便看到并发的效果。

有了 Worker 接口的实现，我们就可以看一下 main 函数内部的代码了，如代码清单7-36所示。

代码清单7-36　 work /main/main.go：第33行到第64行

33 // main是所有Go程序的入口
34 func main() {
35　　 // 使用两个goroutine来创建工作池
36　　 p := work.New(2)
37
38　　 var wg sync.WaitGroup
39　　 wg.Add(100 * len(names))
40
41　　 for i := 0; i < 100; i++ {
42　　　　 // 迭代names切片
43　　　　 for _, name := range names {
44　　　　　　 // 创建一个namePrinter并提供
45　　　　　　 // 指定的名字
46　　　　　　 np := namePrinter{
47　　　　　　　　 name: name,
48　　　　　　 }
49
50　　　　　　 go func() {
51　　　　　　　　 // 将任务提交执行。当Run返回时
52　　　　　　　　 // 我们就知道任务已经处理完成
53　　　　　　　　 p.Run(&np)
54　　　　　　　　 wg.Done()
55　　　　　　 }()
56　　　　 }
57　　 }
58
59　　 wg.Wait()
60
61　　 // 让工作池停止工作，等待所有现有的
62　　 // 工作完成
63　　 p.Shutdown()
64 }

在代码清单7-36第36行，调用 work 包里的 New 函数创建一个工作池。这个调用传入的参数是2，表示这个工作池只会包含两个执行任务的goroutine。在第38行和第39行，声明了一个 WaitGroup ，并初始化为要执行任务的goroutine数。在这个例子里， names 切片里的每个名字都会创建100个goroutine来提交任务。这样就会有一堆goroutine互相竞争，将任务提交到池里。

在第41行到第43行，内部和外部的 for 循环用来声明并创建所有的goroutine。每次内部循环都会创建一个 namePrinter 类型的值，并提供一个用来打印的名字。之后，在第50行，声明了一个匿名函数，并创建一个goroutine执行这个函数。这个goroutine会调用工作池的 Run 方法，将 namePrinter 的值提交到池里。一旦工作池里的goroutine接收到这个值， Run 方法就会返回。这也会导致goroutine将 WaitGroup 的计数递减，并终止goroutine。

一旦所有的goroutine都创建完成， main 函数就会调用 WaitGroup 的 Wait 方法。这个调用会等待所有创建的goroutine提交它们的工作。一旦 Wait 返回，就会调用工作池的 Shutdown 方法来关闭工作池。 Shutdown 方法直到所有的工作都做完才会返回。在这个例子里，最多只会等待两个工作的完成。